package com.zhang.first.day03;

import org.apache.flink.api.common.functions.Partitioner;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @title: 自定义物理分区
 * @author: zhang
 * @date: 2022/1/16 10:44
 */
public class Example1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env
                .fromElements(11, 22, 33, 44,22,33).setParallelism(1)
                .keyBy(r -> r)
                .print("keyBy:").setParallelism(2);


        env
                .fromElements(1, 2, 3, 4).setParallelism(1)
                .partitionCustom(
                        //指定将某个key的数据发送到哪一个物理分区
                        new Partitioner<Integer>() {
                                     @Override
                                     public int partition(Integer key, int numPartitions) {
                                         return key;
                                     }
                                 },
                        //为每个数指定key
                        //每来一条数据就指定一个key，
                        new KeySelector<Integer, Integer>() {
                            @Override
                            public Integer getKey(Integer value) throws Exception {
                                //按照奇偶行分组
                                return value % 2;
                            }
                        })
                .print("自定义物理分区:").setParallelism(2);

        env.execute();

    }
}
